package com.v.im.websocketservice.netty.client;

import com.v.im.commonmodule.utils.zkclient.balance.client.LoadBalanceImplProvider;
import com.v.im.commonmodule.utils.zkclient.balance.server.ServerData;
import com.v.im.websocketservice.config.ZkConnectConfig;
import com.v.im.websocketservice.netty.handle.NettyClientInitializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

@Slf4j
@EqualsAndHashCode
@Data
@Service
public class NettyClient implements INettyClient {
    /**
     * s
     * 各自客户端的channel
     */
    private Channel channel;

    private EventLoopGroup group = new NioEventLoopGroup();

    @Autowired
    private ZkConnectConfig zkConnectConfig;

    /**
     * 失败的请求
     */
    private transient Queue<Object> failMessages = new ConcurrentLinkedQueue<>();

    /**
     * 是否初始化过
     */
    private volatile boolean initialized = Boolean.FALSE;

    /**
     * 是否存活
     */
    private volatile boolean isLive = Boolean.FALSE;


    private void connect() throws InterruptedException {
        channel = configBootstrap().connect().sync().channel();
    }

    private Bootstrap configBootstrap() {
        // 获取候选节点 并 负载均衡算出 该用哪一个server
        LoadBalanceImplProvider loadBalanceImplProvider = zkConnectConfig.getBalanceProvider();
        List<ServerData> candidateServerDataList = loadBalanceImplProvider.getBalanceItems();
        ServerData serverData = loadBalanceImplProvider.balanceAlgorithm(candidateServerDataList);
        log.info("正在连接 "+serverData.getHost()+":"+serverData.getPort()+" im 服务器");
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.remoteAddress(serverData.getHost(), serverData.getPort());
        bootstrap.group(group).channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.handler(new NettyClientInitializer());
        return bootstrap;
    }


    @Override
    public void init() {
        try {
            connect();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            initialized = Boolean.TRUE;
        }

    }

    @Override
    public void destroy() {
        try {
            group.shutdownGracefully().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //log.info("关闭Netty");
    }


    @Override
    public void send(Object object) {
        init();
        if (!initialized) {
            //log.warn("netty client not init, message: {}", toString());
            failMessages.add(object);
            return;
        }
        if (channel == null) {
            //log.warn("channel  is empty! message: {}", toString());
            failMessages.add(object);
            return;
        }
        channel.writeAndFlush(object).addListener(channelFuture -> {
            if (!channelFuture.isSuccess()) {
                failMessages.add(failMessages);
                return;
            }
        });
    }

    @Override
    public void retrySendAll() {
        if (failMessages.size() > 0) {
            while (true) {
                try {
                    send(failMessages.remove());
                } catch (NoSuchElementException ex) {
                    break;
                }
            }
        }
    }
}
